package com.doit.day02;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.sql.*;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.Properties;


/**
 * 1.查数据源有没有问题    kafka  ==》 mysql
 * 2.代码有没有问题
 */

public class _02_Data2Mysql {
    public static void main(String[] args) throws SQLException {
        //获取一个mysql的jdbc连接对象
        Connection connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/test01", "root", "123456");
        //mysqljdbc默认事务是自动开启的
        connection.setAutoCommit(false);

        //预编译sql
        PreparedStatement pps = connection.prepareStatement("insert into user_info values(?,?,?,?)");

        PreparedStatement offset = connection.prepareStatement("insert into tp_offset values(?,?,?,?) on DUPLICATE key update offset = ? ");

        PreparedStatement getOffset = connection.prepareStatement("select offset from tp_offset where group_id = ? and topic = ? and `partition` = ?");


        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "linux01:9092");
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "G01");
        //自动提交偏移量
        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
//        props.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"3000");
        //指定从哪里开始消费
        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        //创建消费者对象
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        //订阅主题
        consumer.subscribe(Arrays.asList("user2"), new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
            }
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                //得到这个消费者在在均衡之后会消费哪些分区的数据
                try {
                    for (TopicPartition partition : partitions) {
                        String topic = partition.topic();
                        int partition1 = partition.partition();
                        //看此时消费的是哪一个分区的数据，就要将该分区的偏移量获取到，从下一条开始消费
                        //获取offset
                        getOffset.setString(1,"G01");
                        getOffset.setString(2,topic);
                        getOffset.setInt(3,partition1);
                        ResultSet resultSet = getOffset.executeQuery();
                        if (resultSet.next()){
                            long off = resultSet.getLong(1);
                            consumer.seek(partition,off+1);
                        }else {
                            consumer.seek(partition,0);
                        }
                    }
                } catch (SQLException e) {
                    System.out.println(e);
                }
            }
        });

      /*  TopicPartition topicPartition0 = new TopicPartition("user2", 0);
        TopicPartition topicPartition1 = new TopicPartition("user2", 1);
        TopicPartition topicPartition2 = new TopicPartition("user2", 2);

        consumer.assign(Arrays.asList(topicPartition0, topicPartition1, topicPartition2));

        getOffset.setString(1, "G01");
        getOffset.setString(2, "user2");
        getOffset.setInt(3, 0);
        ResultSet resultSet = getOffset.executeQuery();
        if (resultSet.next()) {
            long off = resultSet.getLong(1);
            consumer.seek(topicPartition0, off + 1);
        }else{
            consumer.seek(topicPartition0, 0);
        }

        getOffset.setString(1, "G01");
        getOffset.setString(2, "user2");
        getOffset.setInt(3, 1);
        ResultSet resultSet1 = getOffset.executeQuery();

        if (resultSet1.next()) {
            long off = resultSet1.getLong(1);
            consumer.seek(topicPartition1, off + 1);
        }else{
            consumer.seek(topicPartition1, 0);
        }

        getOffset.setString(1, "G01");
        getOffset.setString(2, "user2");
        getOffset.setInt(3, 2);
        ResultSet resultSet2 = getOffset.executeQuery();
        if (resultSet2.next()) {
            long off = resultSet2.getLong(1);
            consumer.seek(topicPartition2, off + 1);
        }else{
            consumer.seek(topicPartition2, 0);
        }*/


        while (true) {
            //拉取数据
            ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(Integer.MAX_VALUE));
            for (ConsumerRecord<String, String> consumerRecord : poll) {

                try {
                    long offset1 = consumerRecord.offset();
                    String topic = consumerRecord.topic();
                    int partition = consumerRecord.partition();
                    offset.setString(1, "G01");
                    offset.setString(2, topic);
                    offset.setInt(3, partition);
                    offset.setLong(4, offset1);
                    offset.setLong(5, offset1);
                    offset.execute();

                    String json = consumerRecord.value();
                    Person person = JSON.parseObject(json, Person.class);
                    //要将数据写入到mysql数据库中
                    pps.setInt(1, person.getId());
                    pps.setString(2, person.getName());
                    pps.setInt(3, person.getAge());
                    pps.setString(4, person.getGender());
                    pps.execute();

                    if (person.getId() == 1000){
                        throw  new Exception("我又来了！！！");
                    }


                    //提交事务
                    connection.commit();
                } catch (Exception e) {
                    //有异常了，回滚事务
                    System.out.println(e);
                    connection.rollback();
                }

            }
        }
    }
}
